package com.ymm.spark.spark.product;

import com.alibaba.fastjson.JSONObject;
import com.ymm.spark.config.ConfigurationManager;
import com.ymm.spark.constant.Constants;
import com.ymm.spark.dao.ITaskDAO;
import com.ymm.spark.dao.factory.DAOFactory;
import com.ymm.spark.domain.Task;
import com.ymm.spark.util.ParamUtils;
import com.ymm.spark.util.SparkUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author zhangj
 * @date 2019/12/8
 */
public class AreaTop3ProductSpark {
	public static void main(String[] args) {
		//创建sparkConf
		SparkConf conf = new SparkConf().setAppName(Constants.SPARK_APP_NAME_PRODUCT);
		SparkUtils.setMaster(conf);

		//构建spark上下文
		JavaSparkContext sc = new JavaSparkContext(conf);
		SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());

		//准备模拟数据
		SparkUtils.mockData(sc, sqlContext);

		//获得命令行传入的taskid,查询对应的任务参数
		final ITaskDAO taskDAO = DAOFactory.getTaskDAO();
		final Long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PRODUCT);
		final Task task = taskDAO.findById(taskid);
		final JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());
		final String startDate = ParamUtils.getParam(taskParam, Constants.PARAM_START_DATE);
		final String endDate = ParamUtils.getParam(taskParam, Constants.PARAM_END_DATE);

		//查询用户指定日期范围内的点击行为数据
		//技术点1:Hive数据源的使用
		JavaPairRDD<Long, Row> cityid2clickActionRDD = getcityid2ClickActionRDDByDate(sqlContext, startDate, endDate);

		//从MYSQL中查询城市信息
		//技术点2:异构数据源之MYSQL的使用
		JavaPairRDD<Long, Row> cityid2cityInfoRDD = getcityid2CityInfoRDD(sqlContext);

		// 生成点击商品基础信息临时表
		// 技术点3：将RDD转换为DataFrame，并注册临时表
		generateTempClickProductBasicTable(sqlContext, cityid2clickActionRDD, cityid2cityInfoRDD);

		sc.close();
	}

	private static void generateTempClickProductBasicTable(SQLContext sqlContext,
	                                                       JavaPairRDD<Long, Row> cityid2clickActionRDD,
	                                                       JavaPairRDD<Long, Row> cityid2cityInfoRDD) {
		//进行join,进行点击行为数据和城市数据的关联
		final JavaPairRDD<Long, Tuple2<Row, Row>> joinedRDD = cityid2clickActionRDD.join(cityid2cityInfoRDD);

		//将上面的JavaPairRDD,转换成一个JavaRDD<Row>,才能将RDD转换成DataFrame
		final JavaRDD<Row> mappedRDD = joinedRDD.map(new Function<Tuple2<Long, Tuple2<Row, Row>>, Row>() {
			@Override
			public Row call(Tuple2<Long, Tuple2<Row, Row>> tuple) throws Exception {
				long cityid = tuple._1;
				Row clickAction = tuple._2._1;
				Row cityInfo = tuple._2._2;

				long productid = clickAction.getLong(1);
				String cityName = cityInfo.getString(1);
				String area = cityInfo.getString(2);
				return RowFactory.create(cityid, cityName, area, productid);
			}
		});

		//将JavaRDD<Row>转换成DataFrame
		List<StructField> structFields = new ArrayList<StructField>();
		structFields.add(DataTypes.createStructField("city_id", DataTypes.LongType, true));
		structFields.add(DataTypes.createStructField("city_name", DataTypes.StringType, true));
		structFields.add(DataTypes.createStructField("area", DataTypes.StringType, true));
		structFields.add(DataTypes.createStructField("product_id", DataTypes.LongType, true));
		final StructType schema = DataTypes.createStructType(structFields);
		final DataFrame df = sqlContext.createDataFrame(mappedRDD, schema);

		//将DataFrame中的数据,注册成临时表
		df.registerTempTable("tmp_clk_prod_basic");
	}

	private static JavaPairRDD<Long, Row> getcityid2CityInfoRDD(SQLContext sqlContext) {
		//构建mysql连接配置信息
		boolean local = ConfigurationManager.getBoolean(Constants.SPARK_LOCAL);
		String url = null;
		if (local) {
			url = ConfigurationManager.getProperty(Constants.JDBC_URL);
		} else {
			url = ConfigurationManager.getProperty(Constants.JDBC_URL_PROD);
		}
		Map<String, String> options = new HashMap<String, String>();
		options.put("url", url);
		options.put("dbtable", "city_info");

		//通过SQLContext从MYSQL中查询数据
		final DataFrame cityInfoDF = sqlContext.read().format("jdbc").options(options).load();

		//返回RDD
		final JavaRDD<Row> cityInfoRDD = cityInfoDF.javaRDD();

		final JavaPairRDD<Long, Row> cityid2cityInfoRDD = cityInfoRDD.mapToPair(new PairFunction<Row, Long, Row>() {
			@Override
			public Tuple2<Long, Row> call(Row row) throws Exception {
				final long cityid = row.getLong(0);
				return new Tuple2<Long, Row>(cityid, row);
			}
		});
		return cityid2cityInfoRDD;
	}

	private static JavaPairRDD<Long, Row> getcityid2ClickActionRDDByDate(SQLContext sqlContext, String startDate, String endDate) {
		//从user_visit_action中,查询用户访问行为数据
		//第一个限定:click_product_id,限定为不为空的访问行为,那么就代表着点击行为
		//第二个限定:在用户指定的日期范围内的数据
		String sql = "SELECT city_id,click_product_id product_id FROM user_visit_action " +
				"WHERE click_product_id IS NOT NULL " +
				"AND click_product_id != 'NULL' " +
				"AND click_product_id != 'null' " +
				"AND action_time >= '" + startDate + "' " +
				"AND action_time <= '" + endDate + "'";
		final DataFrame dataFrame = sqlContext.sql(sql);
		final JavaRDD<Row> clickActionRDD = dataFrame.javaRDD();
		final JavaPairRDD<Long, Row> cityid2clickActionRDD = clickActionRDD.mapToPair(new PairFunction<Row, Long, Row>() {
			@Override
			public Tuple2<Long, Row> call(Row row) throws Exception {
				long cityid = row.getLong(0);
				return new Tuple2<Long, Row>(cityid, row);
			}
		});
		return cityid2clickActionRDD;
	}
}
